package net.sh.dpinfrastructure.support.spring.configuration;

import java.net.InetAddress;
import java.util.concurrent.locks.ReentrantLock;

import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.Primary;

import net.sh.dpinfrastructure.config.DPConfig;
import net.sh.dpinfrastructure.search.DPElasticsearch;
import net.sh.dpinfrastructure.search.DPSearch;

@Configuration
@Import(DPConfigConfiguration.class)
public class DPElasticSearchConfiguration {
    
    private Logger logger = LoggerFactory.getLogger(DPElasticSearchConfiguration.class);
    
    /**
     * 
     * 配置 {@link TransportClient}<br/> 
     * 
     * @param config
     * @return
     *
     * @author shaohui
     * @date 2017年2月11日下午4:10:18
     * @since JDK 1.8
     */
    @Bean(destroyMethod="close", name="defaultElasticsearch")
    @Primary
    public TransportClient getTransportClient(DPConfig config) {
        String clusterName = config.get("module.search.elasticsearch.cluster.name", "dataplatform");
        Settings settings = Settings.builder()
                .put("cluster.name", clusterName)
                .put("client.transport.sniff", true)
                .build();
        @SuppressWarnings("resource")
        TransportClient client = new PreBuiltTransportClient(settings);
        // elasticsearch 节点
        String nodes = config.get("module.search.elasticsearch.nodes");
        String[] hostAndPorts = nodes.split(";");
        for (String hostAndPortStr : hostAndPorts) {
            String[] hostAndPort = hostAndPortStr.split(":");
            String host = hostAndPort[0];
            int port = 9300; // 默认端口
            if (hostAndPort.length > 1) {
                port = Integer.parseInt(hostAndPort[1]);
            }
            try {
                client.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), port));
            } catch (Exception e) {
                logger.error("连接搜索服务器失败", e);
                throw new RuntimeException("连接搜索服务器失败");
            }
        }
        logger.info("elasticsearch connected");
        return client;
    }
    
    /**
     * 
     * 配置 {@link DPElasticsearch}<br/> 
     * 
     * @param client
     * @return
     *
     * @author shaohui
     * @date 2017年2月11日下午4:12:31
     * @since JDK 1.8
     */
    @Bean("elasticsearch")
    @Primary
    public DPSearch getDPElasticsearch(TransportClient client) {
        return new DPElasticsearch(client);
    }
}
